package com.bingxu.flink.action;

import com.bingxu.flink.bean.UserBehavior;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author :BingXu
 * @description :TODO
 * @date :2021/8/13 8:24
 * @modifier :
 */

public class PV_Count {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.BATCH);
        DataStreamSource<String> sourceDS = env.readTextFile("D:\\Source\\JavaProject\\flinkdemo\\input\\UserBehavior.csv");

        SingleOutputStreamOperator<UserBehavior> userBehaviorDS = sourceDS.map(new MapFunction<String, UserBehavior>() {
            @Override
            public UserBehavior map(String s) throws Exception {
                String[] fields = s.split(",");
                return new UserBehavior(Long.parseLong(fields[0]),
                        Long.parseLong(fields[1]),
                        Integer.parseInt(fields[2]),
                        fields[3],
                        Long.parseLong(fields[4])
                );
            }
        });
        userBehaviorDS
                .filter( userBehavior -> "pv".equals(userBehavior.getBehavior()))
                .map(new MapFunction<UserBehavior, Tuple2<String,Integer>>() {
                    @Override
                    public Tuple2<String, Integer> map(UserBehavior userBehavior) throws Exception {
                        return Tuple2.of(userBehavior.getBehavior(),1);
                    }
                })
                .keyBy((KeySelector<Tuple2<String, Integer>, Object>) stringIntegerTuple2 -> stringIntegerTuple2.f0)
                .sum(1)
                .print();

        env.execute();
    }
}
